package jly521.com.graphx

/**
  * 定义用户(id,(name,age)),粉丝关系Edge(idFans,idStar,degree)
  * 统计比自己年纪大的粉丝数及其平均年龄
  *
  * @date 2019/1/2  16:36
  * @author Jly
  */
import org.apache.spark.SparkContext
import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

object myGraphX {

  def main(args: Array[String]) {

    val sparkConf = new SparkConf().setAppName("myGraphPractice").
      setMaster("local[2]")
    val sc = new SparkContext(sparkConf)

    // 顶点RDD[顶点的id,顶点的属性值]
    val users: RDD[(VertexId, (String, Int))] = sc.parallelize(Array((4L, ("David", 18)),
      (1L, ("Alice", 28)), (6L, ("Fran", 40)), (3L, ("Charlie", 30)), (2L, ("Bob", 70)), (5L, ("Ed", 55))))

    val relationships: RDD[Edge[Int]] = sc.parallelize(Array(Edge(4L, 2L, 2),
      Edge(2L, 1L, 7), Edge(4, 5, 8), Edge(2, 4, 2), Edge(5, 6, 3), Edge(3, 2, 4),
      Edge(6, 1, 2), Edge(3, 6, 3), Edge(6, 2, 8), Edge(4, 1, 1), Edge(6, 4, 3)))
    // 默认（缺失）用户
    val defaultUser = ("John Doe", 0)

    //使用RDDs建立一个Graph（有许多建立Graph的数据来源和方法，后面会详细介绍）
    val graph = Graph(users, relationships, defaultUser)

    //定义一个相邻聚合，统计比自己年纪大的粉丝数（count）及其平均年龄（totalAge/count)
    val olderFollowers = graph.aggregateMessages[(Int, Int)](
      triplet => {
        if (triplet.srcAttr._2 > triplet.dstAttr._2) {
          triplet.sendToDst((1, triplet.srcAttr._2))
        }
      },
      (a, b) => (a._1 + b._1, a._2 + b._2), //(2)相当于Reduce函数，a，b各代表一个元组（count，Age）
      TripletFields.All) //(3)可选项,TripletFields.All/Src/Dst

    //计算平均年龄
    val averageOfOlderFollowers = olderFollowers.mapValues(
      (id, value) => value match {
        case (count, totalAge) => (count, totalAge / count)
      })

    averageOfOlderFollowers.foreach(println)
  }
}

